package com.telit.communicate.communicate;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.telit.communicate.constant.CommunicateVariable;
import com.telit.communicate.constant.MsgSource;
import com.telit.communicate.constant.SystemName;
import com.telit.communicate.dto.MsgInfoDTO;
import com.telit.communicate.dto.MsgProcessPathDTO;
import com.telit.communicate.dto.UpAndDownLineDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;


/**
 * @author zhouyu
 * @description 消息处理工厂类
 */
@Component
@Slf4j
public class MessageProcessFactory {

    @Autowired
    @Qualifier("balancedRestTemplate")
    private RestTemplate restTemplate;


    /**
     * 启动消息处理线程
     *
     * @param systemName 待分发系统
     * @param msgSource  消息来源
     */
    public void startMessageProcess(SystemName systemName, MsgSource msgSource) {
        MsgProcessPathDTO msgProcessPathDTO = MsgProcessPathCache.getProcessPath(systemName, msgSource);
        if (msgProcessPathDTO == null) {
            return;
        }
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat(systemName.name() + msgSource.name() + "-Thread-%d").build();
        ExecutorService executorService = Executors.newSingleThreadExecutor(namedThreadFactory);
        executorService.execute(() -> {
            CommunicateVariable.threadHandler.put(systemName.name() + msgSource.name(), true);
            CommunicateVariable.enableReceiveMsg.put(systemName.name() + msgSource.name(), true);
            while (CommunicateVariable.enableReceiveMsg.get(systemName.name() + msgSource.name())!=null) {
                if (!CommunicateVariable.threadHandler.get(systemName.name() + msgSource.name())) {
                    CommunicateVariable.threadHandler.remove(systemName.name() + msgSource.name());
                    break;
                }
                BlockingQueue<Object> msgInfoQueue = MsgInstanceCache.getMsgInfoQueue(msgSource, systemName);
                messageProcess(msgProcessPathDTO, msgInfoQueue);
            }
        });
        //线程池句柄存入缓存
        CommunicateVariable.threadPoolHandler.put(systemName.name() + msgSource.name(), executorService);
    }


    /**
     * 钩子方法，消息处理线程异常结束进行收尾操作
     *
     * @param systemName 待分发系统
     * @param msgSource  消息来源
     */
    public void startShutdownHookProcess(SystemName systemName, MsgSource msgSource) {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (!CommunicateVariable.enableReceiveMsg.get(systemName.name() + msgSource.name())) {
                log.info("message process server stopped,can not receive message....");
                CommunicateVariable.enableReceiveMsg.remove(systemName.name() + msgSource.name());
            }
            MsgProcessPathDTO msgProcessPathDTO = MsgProcessPathCache.getProcessPath(systemName, msgSource);
            if (msgProcessPathDTO == null) {
                return;
            }
            BlockingQueue<Object> unProcessMsgInfoQueue = MsgInstanceCache.getMsgInfoQueue(msgSource, systemName);
            while (!unProcessMsgInfoQueue.isEmpty()) {
                messageProcess(msgProcessPathDTO, unProcessMsgInfoQueue);
            }

        }));

    }


    /**
     * 消息处理
     *
     * @param msgProcessPathDTO 消息处理路径
     * @param msgInfoQueue      消息存放队列
     */
    private void messageProcess(MsgProcessPathDTO msgProcessPathDTO, BlockingQueue<Object> msgInfoQueue) {
        try {
            Object object = msgInfoQueue.take();//为空时线程会阻塞 而不是返回null 不会导致cpu100%
            if (object instanceof MsgInfoDTO) {
                restTemplate.postForEntity(msgProcessPathDTO.getMessageProcessPath(), object, Object.class);
            } else if (object instanceof UpAndDownLineDTO) {
                restTemplate.postForEntity(msgProcessPathDTO.getUpAndDownProcessPath(), object, Object.class);
            }
        } catch (Exception ignored) {

        }
    }
}
